% median.erl
%
% Handles computing global median using gossip
%
-module(median).
-export([merge/2, addFragmentsToMedianStruct/3, new/0, start/4, getList/1, getSum/1, getCount/1, getFragmentIds/1]).

%Struct is a tuple that contains Count,  group of 5 medians, and medians of medians
new() ->
  {0, [], []}.

% start function used to do gossip
start(_Id, Neighbors, MedianStruct, Dict) ->
  random:seed(now()),
  % iterates for number of steps
  TotalIter = ceiling(math:log(length(Neighbors)))+20,
  NewMedianStruct = gMedian_helper(_Id, Neighbors, MedianStruct, Dict, 0, TotalIter),
  io:format("~p's final median: ~p~n", [_Id, getList(NewMedianStruct)]),
  ok.

% main gossip loop, at each step it gets a new node and sends it the median list structure
gMedian_helper(_Id, Neighbors, MedianStruct, Dict, Created, TotalIter) when TotalIter+1 == Created ->
	MedianStruct;
gMedian_helper(_Id, Neighbors, MedianStruct, Dict, Created, TotalIter) ->
   case gossipprob:getGossipNode(Neighbors) of
      {ok, Node} ->
            Node ! {median, MedianStruct, self()};
        {error, _Msg} ->
        ok
    end,
    gMedian_helper(_Id, Neighbors, processQueue(MedianStruct), Dict, Created+1, TotalIter).

% listens for messages from other nodes so it can update its own
processQueue(MedianStruct) ->
	receive
		ReceivedMedianStruct ->  
                     processQueue(ReceivedMedianStruct)  
	after
		1000 -> MedianStruct
	end.

% combines received median struct with existing struct. 
merge(MedianStruct1, MedianStruct2) ->
    Medians1 = element(2, MedianStruct1),
    Medians2 = element(2, MedianStruct2),  
    Medians = median_help(Medians2, Medians1),
    Count = length(Medians),
    MedianOfMedians1 = element(3, MedianStruct1),
    MedianOfMedians2 = element(3, MedianStruct2),
    MedianOfMedians3 = medianOfMedian_help(MedianOfMedians2, MedianOfMedians1),
   
    case Count >= 5 of      
        true ->
            Len = Count - 5,
            NewList = lists:sublist(Medians, 1, 5),
            case Len > 0 of
                true ->
                    RestOfList = lists:sublist(Medians, 6, Len);
                false -> 
                    RestOfList = []
            end,       
            SortNewList = lists:sort(NewList),
            MedianNth = lists:nth(3, SortNewList),
            case lists:member(MedianNth, MedianOfMedians3) of
                false ->
                    MedianOfMedians4 = lists:append(MedianOfMedians3, [MedianNth]);
                true ->
                    MedianOfMedians4 = MedianOfMedians3
            end,
            NewMedian = RestOfList,
            Count4 = length(RestOfList);
        false ->
            NewMedian = Medians,
            MedianOfMedians4 = MedianOfMedians3,
            Count4 = Count
    end,  
    
    NewMedianStruct = {Count4, NewMedian, MedianOfMedians4}.

% add median from list of numbers in a file fragment
addFragmentsToMedianStruct([], _Dict, MedianStruct) ->
  MedianStruct;
addFragmentsToMedianStruct(Keys, Dict, MedianStruct) ->
    FragId = hd(Keys),
    case hasFragmentId(FragId, MedianStruct) of
        false ->
            Count = element(1, MedianStruct),
            MedianOfMedians = element(3, MedianStruct),
            MiddleMedians = element(2, MedianStruct),                    
            case Count >= 5 of           
                true ->                 
                    Len = Count - 5,
                    NewList = lists:sublist(MiddleMedians, 1, 5),
                    case Len > 0 of
                        true ->
                            RestOfList = lists:sublist(MiddleMedians, 6, Count);
                        false -> 
                            RestOfList = []
                    end,
                    SortNewList = lists:sort(NewList),
                    MedianNth = lists:nth(3, SortNewList),
                    case lists:member(MedianNth, MedianOfMedians) of
                        false ->
                            NewMedianOfMedians = lists:append(MedianOfMedians, [MedianNth]);
                        true ->
                            NewMedianOfMedians = MedianOfMedians
                    end,
                    NewMedians = RestOfList,
                    Count2 = length(RestOfList);
                false ->
                    NewMedians = MiddleMedians,
                    NewMedianOfMedians = MedianOfMedians,
                    Count2 = Count                                     
            end,
            FloatList = dict:fetch(FragId, Dict),
            FloatSortedList = lists:sort(FloatList),
            FloatMedian = lists:nth(3, FloatSortedList),
            case lists:member(FloatMedian, NewMedians) of
                false ->
                    NewMedian2 = lists:append(NewMedians, [FloatMedian]);
                true ->
                    NewMedian2 = NewMedians
            end,
            
            NewMedianStruct = {Count2, NewMedian2, NewMedianOfMedians};             
        true ->
            NewMedianStruct = MedianStruct
    end,
    addFragmentsToMedianStruct(tl(Keys), Dict, NewMedianStruct).

%get the median from the medianOfMedian list
getList(MedianStruct) ->
    MedianOfMedianList = element(3, MedianStruct),
    if
        length(MedianOfMedianList) == 0 ->
            Retval = undefined;
        true ->
            SortedList = lists:sort(MedianOfMedianList),
            ListLength = length(MedianOfMedianList),
            Half = round(ListLength/2),    
            Median = lists:nth(Half, SortedList),
            Retval = [Median]
    end,
    Retval.

%helper methods
getSum(MedianStruct) ->
  element(1, MedianStruct).
  
getCount(MedianStruct) ->
  element(2, MedianStruct).

hasFragmentId(Id, MedianStruct) ->
  lists:member(Id, getFragmentIds(MedianStruct)).
  
getFragmentIds(MedianStruct) ->
  element(3, MedianStruct).


%merge medianofmedian lists from two different structs
medianOfMedian_help([], MedianOfMedians1) ->
    MedianOfMedians1;
medianOfMedian_help(MedianOfMedians2,MedianOfMedians1) ->
    case lists:member(hd(MedianOfMedians2), MedianOfMedians1) of
        true ->
            medianOfMedian_help(tl(MedianOfMedians2), MedianOfMedians1);
        false ->
            medianOfMedian_help(tl(MedianOfMedians2),[hd(MedianOfMedians2) | MedianOfMedians1])
    end.

%merge median lists from two different structs
median_help([], MedianList) ->
    MedianList;
median_help(FloatList, MedianList) ->
    case lists:member(hd(FloatList), MedianList) of
        true ->
            median_help(tl(FloatList), MedianList);
        false ->
            median_help(tl(FloatList), [hd(FloatList) | MedianList])
    end.

% floor and ceiling functions
floor(X) ->
    T = erlang:trunc(X),
    case (X - T) of
        Neg when Neg < 0 -> T - 1;
        Pos when Pos > 0 -> T;
        _ -> T
    end.

ceiling(X) ->
    T = erlang:trunc(X),
    case (X - T) of
        Neg when Neg < 0 -> T;
        Pos when Pos > 0 -> T + 1;
        _ -> T
    end.
